package com.huatai.bi.rabbitmq;

import com.alibaba.cloud.nacos.NacosDiscoveryProperties;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.utils.NetUtils;
import com.alibaba.nacos.client.naming.NacosNamingService;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.huatai.bi.constant.RedisConstant;
import com.huatai.bi.entity.MqClientNumberEntity;
import com.huatai.bi.entity.MqMqttClientEntity;
import com.huatai.bi.mqtt.MqttClientConnect;
import com.huatai.bi.service.MqClientNumberService;
import com.huatai.bi.service.MqMqttClientService;
import com.huatai.bi.websocket.MessageResult;
import com.huatai.bi.websocket.WebSocket;
import com.huatai.common.util.CollectionUtil;
import com.huatai.common.util.ObjectUtil;
import com.huatai.common.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.websocket.EncodeException;
import javax.websocket.Session;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @program: ganghua
 * @description: 监听rabbitmq消息
 * @author: songxiude
 * @create: 2024-09-05 10:32
 **/
@Slf4j
@Component
public class SpringRabbitLeistener {

	@Autowired
	private MqMqttClientService mqttClientService;

	@Autowired
	private MqClientNumberService numberService;

	@Value("${server.port}")
	private String port;

	/**
	 * 获取nacos属性配置信息
	 */
	@Autowired
	private NacosDiscoveryProperties nacosDiscoveryProperties;

	@Autowired
	private RedissonClient redissonClient;

	/**
	 * 处理mqtt连接
	 *
	 * @param msg
	 */
	@RabbitListener(queues = "#{fanountQeueueMqtt.name}")
	public void listenFanountQeueueMqtt(String msg) {
		JSONObject jsonObject = JSONObject.parseObject(msg);
		if (ObjectUtil.isEmpty(jsonObject)) {
			return;
		}
		int type = jsonObject.getIntValue("type");
		String ip = NetUtils.localIP() + ":" + port;
		switch (type) {
			//新增
			case 1:
				MqMqttClientEntity entity = jsonObject.getObject("data", MqMqttClientEntity.class);
				String ip_port = numberService.getMinClient(getIpList());
				//如果是最小连接实例创建连接
				if (ip.equals(ip_port)) {
					boolean b = mqttClientService.initMqttServer(entity);
					entity.setStatus(0);
					if (b) {
						entity.setStatus(1);
						entity.setIpPort(ip);
					}
					mqttClientService.updateById(entity);
					//把连接数据更新
					MqClientNumberEntity numberEntity = numberService.getOne(Wrappers.<MqClientNumberEntity>lambdaQuery().eq(MqClientNumberEntity::getIpPort, ip));
					numberEntity.setIsActive(1);
					numberEntity.setClientNumber(numberEntity.getClientNumber() + 1);
					numberService.updateById(numberEntity);
				}
				break;
			//修改
			case 2:
				MqMqttClientEntity clientEntity = jsonObject.getObject("data", MqMqttClientEntity.class);
				//先停掉运行中的链接
				ConcurrentHashMap<String, MqttClientConnect> mqttClients = MqttClientConnect.mqttClients;
				MqttClientConnect clientConnect = mqttClients.get(clientEntity.getClientId());
				if (ObjectUtil.isNotEmpty(clientConnect)) {
					try {
						clientConnect.close();
						mqttClients.remove(clientEntity.getClientId());
						//重新创建连接在本实例中
						boolean b = mqttClientService.initMqttServer(clientEntity);
						if (!b) {
							clientEntity.setStatus(0);
							mqttClientService.updateById(clientEntity);
						}
					} catch (MqttException e) {
						log.error("关闭链接失败");
					}
				}
				break;
			//停止
			case 3:
				MqMqttClientEntity stopEntity = jsonObject.getObject("data", MqMqttClientEntity.class);
				//先停掉运行中的链接
				ConcurrentHashMap<String, MqttClientConnect> stopMqttClients = MqttClientConnect.mqttClients;
				MqttClientConnect stopClientConnect = stopMqttClients.get(stopEntity.getClientId());
				if (ObjectUtil.isNotEmpty(stopClientConnect)) {
					try {
						stopClientConnect.close();
						stopMqttClients.remove(stopEntity.getClientId());
						//把连接数据更新
						MqClientNumberEntity numberEntity = numberService.getOne(Wrappers.<MqClientNumberEntity>lambdaQuery().eq(MqClientNumberEntity::getIpPort, ip));
						numberEntity.setClientNumber(numberEntity.getClientNumber() - 1);
						numberService.updateById(numberEntity);
					} catch (MqttException e) {
						log.error("停止mqtt链接失败");
					}
				}
				break;
			//启动
			case 4:
				MqMqttClientEntity startEntity = jsonObject.getObject("data", MqMqttClientEntity.class);
				String startIpport = numberService.getMinClient(getIpList());
				//找最小连接数实例来创建连接
				if (ip.equals(startIpport)) {
					boolean b = mqttClientService.initMqttServer(startEntity);
					startEntity.setStatus(0);
					startEntity.setEnableFlag(false);
					if (b) {
						startEntity.setStatus(1);
						startEntity.setIpPort(ip);
						startEntity.setEnableFlag(true);
					}
					mqttClientService.updateById(startEntity);
					//把连接数据更新
					MqClientNumberEntity numberEntity = numberService.getOne(Wrappers.<MqClientNumberEntity>lambdaQuery().eq(MqClientNumberEntity::getIpPort, ip));
					numberEntity.setIsActive(1);
					numberEntity.setClientNumber(numberEntity.getClientNumber() + 1);
					numberService.updateById(numberEntity);
				}
				break;
			//删除
			case 5:
				MqMqttClientEntity rmEntity = jsonObject.getObject("data", MqMqttClientEntity.class);
				//先停掉运行中的链接
				ConcurrentHashMap<String, MqttClientConnect> rmMqttClients = MqttClientConnect.mqttClients;
				MqttClientConnect rmClientConnect = rmMqttClients.get(rmEntity.getClientId());
				if (ObjectUtil.isNotEmpty(rmClientConnect)) {
					try {
						rmClientConnect.close();
						rmMqttClients.remove(rmEntity.getClientId());
						//把连接数据更新
						MqClientNumberEntity numberEntity = numberService.getOne(Wrappers.<MqClientNumberEntity>lambdaQuery().eq(MqClientNumberEntity::getIpPort, ip));
						numberEntity.setClientNumber(numberEntity.getClientNumber() - 1);
						numberService.updateById(numberEntity);
					} catch (MqttException e) {
						log.error("断开连接失败");
					}

				}
				break;
			default:
				log.error("不匹配任何条件");
		}
	}


	/**
	 * 实例下线处理转移
	 *
	 * @param msg
	 */
	@RabbitListener(queues = "#{serverQeueueMqtt.name}")
	public void serverQeueueMqtt(String msg) {
		List<MqClientNumberEntity> numberEntities = JSONObject.parseArray(msg, MqClientNumberEntity.class);
		if (CollectionUtil.isEmpty(numberEntities)) {
			return;
		}
		List<String> ips = numberEntities.stream().map(entity -> entity.getIpPort()).collect(Collectors.toList());
		List<MqMqttClientEntity> clientEntities = mqttClientService.list(Wrappers.<MqMqttClientEntity>lambdaQuery()
			.eq(MqMqttClientEntity::getStatus, 0)
			.in(MqMqttClientEntity::getIpPort, ips));
		if (CollectionUtil.isEmpty(clientEntities)) {
			return;
		}
		String ip = NetUtils.localIP();
		for (MqMqttClientEntity clientEntity : clientEntities) {
			RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(RedisConstant.MQTT_CLIENT + clientEntity.getId());
			RLock rLock = readWriteLock.writeLock();
			rLock.lock();
			try {
				//再次确认是否被其他服务连接
				MqMqttClientEntity mqttClient = mqttClientService.getById(clientEntity.getId());
				if (mqttClient.getStatus().intValue() == 0 && mqttClient.getEnableFlag()) {
					boolean b = mqttClientService.initMqttServer(mqttClient);
					if (b) {
						mqttClient.setStatus(1);
						mqttClient.setIpPort(ip + ":" + port);
						mqttClientService.updateById(mqttClient);
					}
				}
			} catch (Exception e) {
				log.error(e.getMessage(), e);
			} finally {
				rLock.unlock();
			}
		}
		//更新连接统计数量
		MqClientNumberEntity numberEntity = numberService.getOne(Wrappers.<MqClientNumberEntity>lambdaQuery().eq(MqClientNumberEntity::getIpPort, ip + ":" + port));
		int size = MqttClientConnect.mqttClients.size();
		numberEntity.setClientNumber(size);
		numberService.saveOrUpdate(numberEntity);
		log.info(ip + ":" + port + "执行了一次mqtt连接更新" + size);
	}


	@RabbitListener(queues = "#{mqttQueueWebsocket.name}")
	public void queueWebsocket(String jsonObject) {
		log.info("收到socket数据信息："+jsonObject);
//		Map<String, MessageResult> map = JSONObject.parseObject(jsonObject,Map.class);
//		Set<Map.Entry<String, MessageResult>> entries = map.entrySet();
		if(StringUtil.isEmpty(jsonObject)){
			return;
		}
		Map<String, Session> electricSocketMap = WebSocket.electricSocketMap;
		log.info("electricSocketMap："+electricSocketMap);
		if(ObjectUtil.isEmpty(electricSocketMap)){
			return;
		}
		MessageResult messageResult = JSONObject.parseObject(jsonObject, MessageResult.class);
		if(ObjectUtil.isEmpty(messageResult)){
			return;
		}
		//System.out.println("在线："+electricSocketMap.size());
		for (Map.Entry<String, Session> entry : electricSocketMap.entrySet()) {
			//WebSocket.sendMsg(entry.getValue(), messageResult);
			if(StringUtil.isNoneBlank(messageResult.getTopic())) {
				String topic = messageResult.getTopic().substring(0, messageResult.getTopic().lastIndexOf("_"));
				if(entry.getKey().contains(topic)){
					WebSocket.sendMsg(entry.getValue(), messageResult);
				}
			}
		}
//		for (Map.Entry<String, MessageResult> entry : entries) {
//			log.info("消息接收者接收到来自【队列一】的消息，消息内容: "+jsonObject);
//			Map<String, Session> electricSocketMap = WebSocket.electricSocketMap;
//			System.out.println("几个："+electricSocketMap.size());
//			System.out.println(entry.getKey()+"的值为"+entry.getValue());
//			if(electricSocketMap.containsKey(entry.getKey())){
//				Object obj = map.get(entry.getKey());
//				MessageResult result = (MessageResult) JSONObject.parseObject(obj.toString(),MessageResult.class);
//				WebSocket.sendMsg(electricSocketMap.get(entry.getKey()),result);
//			}
//		}

	}




	/**
	 * 获取该服务在线总实例
	 *
	 * @return
	 */
	private List<String> getIpList() {
		NacosNamingService nacosNamingService = null;
		List<String> list = new ArrayList<>();
		try {
			nacosNamingService = new NacosNamingService(nacosDiscoveryProperties.getNacosProperties());
			List<Instance> instances = nacosNamingService.getAllInstances("bi");
			for (Instance instance : instances) {
				list.add(instance.getIp() + ":" + instance.getPort());
			}
		} catch (NacosException e) {
			log.error("获取实例列表失败");
			return Collections.emptyList();
		}
		return list;

	}



}
